package com.example.framework.mq.kafka.core.util;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.IdUtil;
import com.example.framework.common.util.json.JsonUtils;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;

@Slf4j
@RequiredArgsConstructor
public class KafkaProducerUtils {

    private static KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
        KafkaProducerUtils.kafkaTemplate = kafkaTemplate;
    }

    @SneakyThrows
    public static void sendMessage(String topic, Object msgBody) {
        log.info("准备发送Kafka消息，topic={}, msg={}", topic, msgBody);
        TimeInterval timer = DateUtil.timer();
        RecordMetadata recordMetadata = kafkaTemplate.send(topic, IdUtil.fastSimpleUUID(), JsonUtils.toJsonString(msgBody)).get().getRecordMetadata();
        log.info("<=== send kafka [end] success cause:[{}], topic:[{}], partition:[{}], offset:[{}]", timer.intervalPretty(), topic, recordMetadata.partition(), recordMetadata.offset());
    }

    @SneakyThrows
    public static void sendMessageAsync(String topic, Object msgBody) {
        log.info("准备发送异步Kafka消息，topic={}, msg={}", topic, msgBody);
        TimeInterval timer = DateUtil.timer();
        kafkaTemplate.send(topic, IdUtil.fastSimpleUUID(), JsonUtils.toJsonString(msgBody))
                .thenAccept(sendResult -> {
                    RecordMetadata recordMetadata = sendResult.getRecordMetadata();
                    log.info("<=== send kafka async [end] success cause:[{}], topic:[{}], partition:[{}], offset:[{}]", timer.intervalPretty(), topic, recordMetadata.partition(), recordMetadata.offset());
                })
                .exceptionally(throwable -> {
                    log.error("发送Kafka消息失败，msg={}, e={}", msgBody, throwable.getMessage());
                    return null;
                });
    }

}
